package com.xneure.utils;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.Charset;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.xneure.utils.BytesUtil.LexicographicalComparerHolder.UnsafeComparer;

import sun.misc.Unsafe;

@SuppressWarnings("restriction")
public class BytesUtil{
	static Logger logger = Logger.getLogger(BytesUtil.class.getName());
	private static final String UTF8_ENCODING = "UTF-8";
	private static final Charset UTF8_CHARSET = Charset.forName(UTF8_ENCODING);
	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
	public static final int SIZEOF_BOOLEAN = Byte.SIZE / Byte.SIZE;
	public static final int SIZEOF_INT = Integer.SIZE / Byte.SIZE;
	public static final int SIZEOF_LONG = Long.SIZE / Byte.SIZE;
	public static final int SIZEOF_SHORT = Short.SIZE / Byte.SIZE;
	private static final boolean UNSAFE_UNALIGNED = UnsafeAvailChecker.unaligned();

	/**
	 * 返回字节数组的长度，如果数组为空则返回0。用于计算大小。
	 *
	 * @param b字节数组，可以为空
	 * @return 0如果b为空，则返回长度
	 */
	final public static int len(byte[] b){
		return b == null ? 0 : b.length;
	}

	/**
	 * 将字节放在指定的字节数组位置。
	 *
	 * @param tgtBytes字节数组
	 * @param tgtOffset在数组中的位置
	 * @param srcBytes数组写入
	 * @param srcOffset源偏移
	 * @param srcLength源长度
	 * @return 递增偏移量
	 */
	public static int putBytes(byte[] tgtBytes,int tgtOffset,byte[] srcBytes,int srcOffset,int srcLength){
		System.arraycopy(srcBytes,srcOffset,tgtBytes,tgtOffset,srcLength);
		return tgtOffset + srcLength;
	}

	/**
	 * 将单个字节写入指定的字节数组位置。
	 *
	 * @param 字节数组
	 * @param 在数组中的偏移位置
	 * @param b字节写出来
	 * @return 递增偏移量
	 */
	public static int putByte(byte[] bytes,int offset,byte b){
		bytes[offset] = b;
		return offset + 1;
	}

	/**
	 * 将ByteBuffer的全部内容添加到字节数组中。ByteBuffer被修改了。
	 *
	 * @param 字节字节数组 在数组中
	 * @param  偏移位置
	 * @param buf ByteBuffer写出来
	 * @返回增量偏移
	 */
	public static int putByteBuffer(byte[] bytes,int offset,ByteBuffer buf){
		int len = buf.remaining();
		buf.get(bytes,offset,len);
		return offset + len;
	}

	/**

	*返回一个新字节数组，从给定的{@代码BUF}复制，从索引0（包含）到极限
	*（排他性），不管当前位置。位置和其他索引参数不变。
	*@ PARAM BUF字节缓冲区
	*@返回字节数组
	*@参见ByteBuffer字节（Gub）
	*/
	public static byte[] toBytes(ByteBuffer buf){
		ByteBuffer dup = buf.duplicate();
		dup.position(0);
		return readBytes(dup);
	}

	private static byte[] readBytes(ByteBuffer buf){
		byte[] result = new byte[buf.remaining()];
		buf.get(result);
		return result;
	}

	/**
	 * @param b Presumed UTF-8 encoded byte array.
	 * @return String made from <code>b</code>
	 */
	public static String toString(final byte[] b){
		if(b == null){
			return null;
		}
		return toString(b,0,b.length);
	}

	/**
	 * Joins two byte arrays together using a separator.
	 * 
	 * @param b1 The first byte array.
	 * @param sep The separator to use.
	 * @param b2 The second byte array.
	 */
	public static String toString(final byte[] b1,String sep,final byte[] b2){
		return toString(b1,0,b1.length) + sep + toString(b2,0,b2.length);
	}

	/**
	 * This method will convert utf8 encoded bytes into a string. If the given byte array is null, this method will
	 * return null.
	 * 
	 * @param b Presumed UTF-8 encoded byte array.
	 * @param off offset into array
	 * @return String made from <code>b</code> or null
	 */
	public static String toString(final byte[] b,int off){
		if(b == null){
			return null;
		}
		int len = b.length - off;
		if(len <= 0){
			return "";
		}
		return new String(b,off,len,UTF8_CHARSET);
	}

	/**
	 * This method will convert utf8 encoded bytes into a string. If the given byte array is null, this method will
	 * return null.
	 *
	 * @param b Presumed UTF-8 encoded byte array.
	 * @param off offset into array
	 * @param len length of utf-8 sequence
	 * @return String made from <code>b</code> or null
	 */
	public static String toString(final byte[] b,int off,int len){
		if(b == null){
			return null;
		}
		if(len == 0){
			return "";
		}
		return new String(b,off,len,UTF8_CHARSET);
	}

	/**
	 * Write a printable representation of a byte array.
	 *
	 * @param b byte array
	 * @return string
	 * @see #toStringBinary(byte[], int, int)
	 */
	public static String toStringBinary(final byte[] b){
		if(b == null)
			return "null";
		return toStringBinary(b,0,b.length);
	}

	/**
	 * Converts the given byte buffer to a printable representation, from the index 0 (inclusive) to the limit
	 * (exclusive), regardless of the current position. The position and the other index parameters are not changed.
	 *
	 * @param buf a byte buffer
	 * @return a string representation of the buffer's binary contents
	 * @see #toBytes(ByteBuffer)
	 * @see #getBytes(ByteBuffer)
	 */
	public static String toStringBinary(ByteBuffer buf){
		if(buf == null)
			return "null";
		if(buf.hasArray()){
			return toStringBinary(buf.array(),buf.arrayOffset(),buf.limit());
		}
		return toStringBinary(toBytes(buf));
	}

	/**
	 * Write a printable representation of a byte array. Non-printable characters are hex escaped in the format \\x%02X,
	 * eg: \x00 \x05 etc
	 *
	 * @param b array to write out
	 * @param off offset to start at
	 * @param len length to write
	 * @return string output
	 */
	public static String toStringBinary(final byte[] b,int off,int len){
		StringBuilder result = new StringBuilder();
		// Just in case we are passed a 'len' that is > buffer length...
		if(off >= b.length)
			return result.toString();
		if(off + len > b.length)
			len = b.length - off;
		for(int i = off;i < off + len;++i){
			int ch = b[i] & 0xFF;
			if((ch >= '0' && ch <= '9') || (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || " `~!@#$%^&*()-_=+[]{}|;:'\",.<>/?".indexOf(ch) >= 0){
				result.append((char)ch);
			}else{
				result.append(String.format("\\x%02X",ch));
			}
		}
		return result.toString();
	}

	private static boolean isHexDigit(char c){
		return (c >= 'A' && c <= 'F') || (c >= '0' && c <= '9');
	}

	/**
	 * Takes a ASCII digit in the range A-F0-9 and returns the corresponding integer/ordinal value.
	 * 
	 * @param ch The hex digit.
	 * @return The converted hex value as a byte.
	 */
	public static byte toBinaryFromHex(byte ch){
		if(ch >= 'A' && ch <= 'F')
			return (byte)((byte)10 + (byte)(ch - 'A'));
		// else
		return (byte)(ch - '0');
	}

	public static byte[] toBytesBinary(String in){
		// this may be bigger than we need, but let's be safe.
		byte[] b = new byte[in.length()];
		int size = 0;
		for(int i = 0;i < in.length();++i){
			char ch = in.charAt(i);
			if(ch == '\\' && in.length() > i + 1 && in.charAt(i + 1) == 'x'){
				// ok, take next 2 hex digits.
				char hd1 = in.charAt(i + 2);
				char hd2 = in.charAt(i + 3);
				// they need to be A-F0-9:
				if(!isHexDigit(hd1) || !isHexDigit(hd2)){
					// bogus escape code, ignore:
					continue;
				}
				// turn hex ASCII digit -> number
				byte d = (byte)((toBinaryFromHex((byte)hd1) << 4) + toBinaryFromHex((byte)hd2));
				b[size++] = d;
				i += 3; // skip 3
			}else{
				b[size++] = (byte)ch;
			}
		}
		// resize:
		byte[] b2 = new byte[size];
		System.arraycopy(b,0,b2,0,size);
		return b2;
	}

	/**
	 * Converts a string to a UTF-8 byte array.
	 * 
	 * @param s string
	 * @return the byte array
	 */
	public static byte[] toBytes(String s){
		return s.getBytes(UTF8_CHARSET);
	}

	/**
	 * Convert a boolean to a byte array. True becomes -1 and false becomes 0.
	 *
	 * @param b value
	 * @return <code>b</code> encoded in a byte array.
	 */
	public static byte[] toBytes(final boolean b){
		return new byte[]{b ? (byte)-1 : (byte)0};
	}

	/**
	 * Reverses {@link #toBytes(boolean)}
	 * 
	 * @param b array
	 * @return True or false.
	 */
	public static boolean toBoolean(final byte[] b){
		if(b.length != 1){
			throw new IllegalArgumentException("Array has wrong size: " + b.length);
		}
		return b[0] != (byte)0;
	}

	/**
	 * Convert a long value to a byte array using big-endian.
	 *
	 * @param val value to convert
	 * @return the byte array
	 */
	public static byte[] toBytes(long val){
		byte[] b = new byte[8];
		for(int i = 7;i > 0;i--){
			b[i] = (byte)val;
			val >>>= 8;
		}
		b[0] = (byte)val;
		return b;
	}

	/**
	 * Converts a byte array to a long value. Reverses {@link #toBytes(long)}
	 * 
	 * @param bytes array
	 * @return the long value
	 */
	public static long toLong(byte[] bytes){
		return toLong(bytes,0,SIZEOF_LONG);
	}

	/**
	 * Converts a byte array to a long value. Assumes there will be {@link #SIZEOF_LONG} bytes available.
	 *
	 * @param bytes bytes
	 * @param offset offset
	 * @return the long value
	 */
	public static long toLong(byte[] bytes,int offset){
		return toLong(bytes,offset,SIZEOF_LONG);
	}

	/**
	 * Converts a byte array to a long value.
	 *
	 * @param bytes array of bytes
	 * @param offset offset into array
	 * @param length length of data (must be {@link #SIZEOF_LONG})
	 * @return the long value
	 * @throws IllegalArgumentException if length is not {@link #SIZEOF_LONG} or if there's not enough room in the array
	 *             at the offset indicated.
	 */
	public static long toLong(byte[] bytes,int offset,final int length){
		if(length != SIZEOF_LONG || offset + length > bytes.length){
			throw explainWrongLengthOrOffset(bytes,offset,length,SIZEOF_LONG);
		}
		if(UNSAFE_UNALIGNED){
			return toLongUnsafe(bytes,offset);
		}else{
			long l = 0;
			for(int i = offset;i < offset + length;i++){
				l <<= 8;
				l ^= bytes[i] & 0xFF;
			}
			return l;
		}
	}

	private static IllegalArgumentException explainWrongLengthOrOffset(final byte[] bytes,final int offset,final int length,final int expectedLength){
		String reason;
		if(length != expectedLength){
			reason = "Wrong length: " + length + ", expected " + expectedLength;
		}else{
			reason = "offset (" + offset + ") + length (" + length + ") exceed the" + " capacity of the array: " + bytes.length;
		}
		return new IllegalArgumentException(reason);
	}

	/**
	 * Presumes float encoded as IEEE 754 floating-point "single format"
	 * 
	 * @param bytes byte array
	 * @return Float made from passed byte array.
	 */
	public static float toFloat(byte[] bytes){
		return toFloat(bytes,0);
	}

	/**
	 * Presumes float encoded as IEEE 754 floating-point "single format"
	 * 
	 * @param bytes array to convert
	 * @param offset offset into array
	 * @return Float made from passed byte array.
	 */
	public static float toFloat(byte[] bytes,int offset){
		int ti = toInt(bytes,offset,SIZEOF_INT);
		return Float.intBitsToFloat(ti);
	}

	/**
	 * @param bytes byte array
	 * @param offset offset to write to
	 * @param f float value
	 * @return New offset in <code>bytes</code>
	 */
	public static int putFloat(byte[] bytes,int offset,float f){
		return putInt(bytes,offset,Float.floatToRawIntBits(f));
	}

	/**
	 * @param f float value
	 * @return the float represented as byte []
	 */
	public static byte[] toBytes(final float f){
		// Encode it as int
		return toBytes(Float.floatToRawIntBits(f));
	}

	/**
	 * @param bytes byte array
	 * @return Return double made from passed bytes.
	 */
	public static double toDouble(final byte[] bytes){
		return toDouble(bytes,0);
	}

	/**
	 * @param bytes byte array
	 * @param offset offset where double is
	 * @return Return double made from passed bytes.
	 */
	public static double toDouble(final byte[] bytes,final int offset){
		return Double.longBitsToDouble(toLong(bytes,offset,SIZEOF_LONG));
	}

	/**
	 * Serialize a double as the IEEE 754 double format output. The resultant array will be 8 bytes long.
	 *
	 * @param d value
	 * @return the double represented as byte []
	 */
	public static byte[] toBytes(final double d){
		// Encode it as a long
		return toBytes(Double.doubleToRawLongBits(d));
	}

	/**
	 * Convert an int value to a byte array. Big-endian. Same as what DataOutputStream.writeInt does.
	 *
	 * @param val value
	 * @return the byte array
	 */
	public static byte[] toBytes(int val){
		byte[] b = new byte[4];
		for(int i = 3;i > 0;i--){
			b[i] = (byte)val;
			val >>>= 8;
		}
		b[0] = (byte)val;
		return b;
	}

	/**
	 * Converts a byte array to an int value
	 * 
	 * @param bytes byte array
	 * @return the int value
	 */
	public static int toInt(byte[] bytes){
		return toInt(bytes,0,SIZEOF_INT);
	}

	/**
	 * Converts a byte array to an int value
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @return the int value
	 */
	public static int toInt(byte[] bytes,int offset){
		return toInt(bytes,offset,SIZEOF_INT);
	}

	/**
	 * Converts a byte array to an int value
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @param length length of int (has to be {@link #SIZEOF_INT})
	 * @return the int value
	 * @throws IllegalArgumentException if length is not {@link #SIZEOF_INT} or if there's not enough room in the array
	 *             at the offset indicated.
	 */
	public static int toInt(byte[] bytes,int offset,final int length){
		if(length != SIZEOF_INT || offset + length > bytes.length){
			throw explainWrongLengthOrOffset(bytes,offset,length,SIZEOF_INT);
		}
		if(UNSAFE_UNALIGNED){
			return toIntUnsafe(bytes,offset);
		}else{
			int n = 0;
			for(int i = offset;i < (offset + length);i++){
				n <<= 8;
				n ^= bytes[i] & 0xFF;
			}
			return n;
		}
	}

	/**
	 * Converts a byte array to an int value (Unsafe version)
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @return the int value
	 */
	public static int toIntUnsafe(byte[] bytes,int offset){
		if(UnsafeComparer.littleEndian){
			return Integer.reverseBytes(UnsafeComparer.theUnsafe.getInt(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET));
		}else{
			return UnsafeComparer.theUnsafe.getInt(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET);
		}
	}

	/**
	 * Converts a byte array to an short value (Unsafe version)
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @return the short value
	 */
	public static short toShortUnsafe(byte[] bytes,int offset){
		if(UnsafeComparer.littleEndian){
			return Short.reverseBytes(UnsafeComparer.theUnsafe.getShort(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET));
		}else{
			return UnsafeComparer.theUnsafe.getShort(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET);
		}
	}

	/**
	 * Converts a byte array to an long value (Unsafe version)
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @return the long value
	 */
	public static long toLongUnsafe(byte[] bytes,int offset){
		if(UnsafeComparer.littleEndian){
			return Long.reverseBytes(UnsafeComparer.theUnsafe.getLong(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET));
		}else{
			return UnsafeComparer.theUnsafe.getLong(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET);
		}
	}

	/**
	 * Converts a byte array to an int value
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @param length how many bytes should be considered for creating int
	 * @return the int value
	 * @throws IllegalArgumentException if there's not enough room in the array at the offset indicated.
	 */
	public static int readAsInt(byte[] bytes,int offset,final int length){
		if(offset + length > bytes.length){
			throw new IllegalArgumentException("offset (" + offset + ") + length (" + length + ") exceed the" + " capacity of the array: " + bytes.length);
		}
		int n = 0;
		for(int i = offset;i < (offset + length);i++){
			n <<= 8;
			n ^= bytes[i] & 0xFF;
		}
		return n;
	}

	/**
	 * Put an int value out to the specified byte array position.
	 * 
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val int to write out
	 * @return incremented offset
	 * @throws IllegalArgumentException if the byte array given doesn't have enough room at the offset specified.
	 */
	public static int putInt(byte[] bytes,int offset,int val){
		if(bytes.length - offset < SIZEOF_INT){
			throw new IllegalArgumentException("Not enough room to put an int at" + " offset " + offset + " in a " + bytes.length + " byte array");
		}
		if(UNSAFE_UNALIGNED){
			return putIntUnsafe(bytes,offset,val);
		}else{
			for(int i = offset + 3;i > offset;i--){
				bytes[i] = (byte)val;
				val >>>= 8;
			}
			bytes[offset] = (byte)val;
			return offset + SIZEOF_INT;
		}
	}

	/**
	 * Put an int value out to the specified byte array position (Unsafe).
	 * 
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val int to write out
	 * @return incremented offset
	 */
	public static int putIntUnsafe(byte[] bytes,int offset,int val){
		if(UnsafeComparer.littleEndian){
			val = Integer.reverseBytes(val);
		}
		UnsafeComparer.theUnsafe.putInt(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET,val);
		return offset + SIZEOF_INT;
	}

	/**
	 * Convert a short value to a byte array of {@link #SIZEOF_SHORT} bytes long.
	 * 
	 * @param val value
	 * @return the byte array
	 */
	public static byte[] toBytes(short val){
		byte[] b = new byte[SIZEOF_SHORT];
		b[1] = (byte)val;
		val >>= 8;
		b[0] = (byte)val;
		return b;
	}

	/**
	 * Converts a byte array to a short value
	 * 
	 * @param bytes byte array
	 * @return the short value
	 */
	public static short toShort(byte[] bytes){
		return toShort(bytes,0,SIZEOF_SHORT);
	}

	/**
	 * Converts a byte array to a short value
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @return the short value
	 */
	public static short toShort(byte[] bytes,int offset){
		return toShort(bytes,offset,SIZEOF_SHORT);
	}

	/**
	 * Converts a byte array to a short value
	 * 
	 * @param bytes byte array
	 * @param offset offset into array
	 * @param length length, has to be {@link #SIZEOF_SHORT}
	 * @return the short value
	 * @throws IllegalArgumentException if length is not {@link #SIZEOF_SHORT} or if there's not enough room in the
	 *             array at the offset indicated.
	 */
	public static short toShort(byte[] bytes,int offset,final int length){
		if(length != SIZEOF_SHORT || offset + length > bytes.length){
			throw explainWrongLengthOrOffset(bytes,offset,length,SIZEOF_SHORT);
		}
		if(UNSAFE_UNALIGNED){
			return toShortUnsafe(bytes,offset);
		}else{
			short n = 0;
			n ^= bytes[offset] & 0xFF;
			n <<= 8;
			n ^= bytes[offset + 1] & 0xFF;
			return n;
		}
	}

	/**
	 * Returns a new byte array, copied from the given {@code buf}, from the position (inclusive) to the limit
	 * (exclusive). The position and the other index parameters are not changed.
	 *
	 * @param buf a byte buffer
	 * @return the byte array
	 * @see #toBytes(ByteBuffer)
	 */
	public static byte[] getBytes(ByteBuffer buf){
		return readBytes(buf.duplicate());
	}

	/**
	 * Put a short value out to the specified byte array position.
	 * 
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val short to write out
	 * @return incremented offset
	 * @throws IllegalArgumentException if the byte array given doesn't have enough room at the offset specified.
	 */
	public static int putShort(byte[] bytes,int offset,short val){
		if(bytes.length - offset < SIZEOF_SHORT){
			throw new IllegalArgumentException("Not enough room to put a short at" + " offset " + offset + " in a " + bytes.length + " byte array");
		}
		if(UNSAFE_UNALIGNED){
			return putShortUnsafe(bytes,offset,val);
		}else{
			bytes[offset + 1] = (byte)val;
			val >>= 8;
			bytes[offset] = (byte)val;
			return offset + SIZEOF_SHORT;
		}
	}

	/**
	 * Put a short value out to the specified byte array position (Unsafe).
	 * 
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val short to write out
	 * @return incremented offset
	 */
	public static int putShortUnsafe(byte[] bytes,int offset,short val){
		if(UnsafeComparer.littleEndian){
			val = Short.reverseBytes(val);
		}
		UnsafeComparer.theUnsafe.putShort(bytes,(long)offset + UnsafeComparer.BYTE_ARRAY_BASE_OFFSET,val);
		return offset + SIZEOF_SHORT;
	}

	/**
	 * Put an int value as short out to the specified byte array position. Only the lower 2 bytes of the short will be
	 * put into the array. The caller of the API need to make sure they will not loose the value by doing so. This is
	 * useful to store an unsigned short which is represented as int in other parts.
	 * 
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val value to write out
	 * @return incremented offset
	 * @throws IllegalArgumentException if the byte array given doesn't have enough room at the offset specified.
	 */
	public static int putAsShort(byte[] bytes,int offset,int val){
		if(bytes.length - offset < SIZEOF_SHORT){
			throw new IllegalArgumentException("Not enough room to put a short at" + " offset " + offset + " in a " + bytes.length + " byte array");
		}
		bytes[offset + 1] = (byte)val;
		val >>= 8;
		bytes[offset] = (byte)val;
		return offset + SIZEOF_SHORT;
	}

	/**
	 * Convert a BigDecimal value to a byte array
	 *
	 * @param val
	 * @return the byte array
	 */
	public static byte[] toBytes(BigDecimal val){
		byte[] valueBytes = val.unscaledValue().toByteArray();
		byte[] result = new byte[valueBytes.length + SIZEOF_INT];
		int offset = putInt(result,0,val.scale());
		putBytes(result,offset,valueBytes,0,valueBytes.length);
		return result;
	}

	/**
	 * Converts a byte array to a BigDecimal
	 *
	 * @param bytes
	 * @return the char value
	 */
	public static BigDecimal toBigDecimal(byte[] bytes){
		return toBigDecimal(bytes,0,bytes.length);
	}

	/**
	 * Converts a byte array to a BigDecimal value
	 *
	 * @param bytes
	 * @param offset
	 * @param length
	 * @return the char value
	 */
	public static BigDecimal toBigDecimal(byte[] bytes,int offset,final int length){
		if(bytes == null || length < SIZEOF_INT + 1 || (offset + length > bytes.length)){
			return null;
		}
		int scale = toInt(bytes,offset);
		byte[] tcBytes = new byte[length - SIZEOF_INT];
		System.arraycopy(bytes,offset + SIZEOF_INT,tcBytes,0,length - SIZEOF_INT);
		return new BigDecimal(new BigInteger(tcBytes),scale);
	}

	/**
	 * Put a BigDecimal value out to the specified byte array position.
	 *
	 * @param bytes the byte array
	 * @param offset position in the array
	 * @param val BigDecimal to write out
	 * @return incremented offset
	 */
	public static int putBigDecimal(byte[] bytes,int offset,BigDecimal val){
		if(bytes == null){
			return offset;
		}
		byte[] valueBytes = val.unscaledValue().toByteArray();
		byte[] result = new byte[valueBytes.length + SIZEOF_INT];
		offset = putInt(result,offset,val.scale());
		return putBytes(result,offset,valueBytes,0,valueBytes.length);
	}

	/**
	 * @param left left operand
	 * @param right right operand
	 * @return 0 if equal, < 0 if left is less than right, etc.
	 */
	public static int compareTo(final byte[] left,final byte[] right){
		return LexicographicalComparerHolder.BEST_COMPARER.compareTo(left,0,left.length,right,0,right.length);
	}

	/**
	 * Lexicographically compare two arrays.
	 *
	 * @param buffer1 left operand
	 * @param buffer2 right operand
	 * @param offset1 Where to start comparing in the left buffer
	 * @param offset2 Where to start comparing in the right buffer
	 * @param length1 How much to compare from the left buffer
	 * @param length2 How much to compare from the right buffer
	 * @return 0 if equal, < 0 if left is less than right, etc.
	 */
	public static int compareTo(byte[] buffer1,int offset1,int length1,byte[] buffer2,int offset2,int length2){
		return LexicographicalComparerHolder.BEST_COMPARER.compareTo(buffer1,offset1,length1,buffer2,offset2,length2);
	}

	interface Comparer<T>{
		int compareTo(T buffer1,int offset1,int length1,T buffer2,int offset2,int length2);
	}

	static Comparer<byte[]> lexicographicalComparerJavaImpl(){
		return LexicographicalComparerHolder.PureJavaComparer.INSTANCE;
	}

	/**
	 * Provides a lexicographical comparer implementation; either a Java implementation or a faster implementation based
	 * on {@link Unsafe}.
	 * <p>
	 * Uses reflection to gracefully fall back to the Java implementation if {@code Unsafe} isn't available.
	 */
	static class LexicographicalComparerHolder{
		static final String UNSAFE_COMPARER_NAME = LexicographicalComparerHolder.class.getName() + "$UnsafeComparer";
		static final Comparer<byte[]> BEST_COMPARER = getBestComparer();

		/**
		 * Returns the Unsafe-using Comparer, or falls back to the pure-Java implementation if unable to do so.
		 */
		static Comparer<byte[]> getBestComparer(){
			try{
				Class<?> theClass = Class.forName(UNSAFE_COMPARER_NAME);
				// yes, UnsafeComparer does implement Comparer<byte[]>
				@SuppressWarnings("unchecked")
				Comparer<byte[]> comparer = (Comparer<byte[]>)theClass.getEnumConstants()[0];
				return comparer;
			}catch(Throwable t){ // ensure we really catch *everything*
				return lexicographicalComparerJavaImpl();
			}
		}

		enum PureJavaComparer implements Comparer<byte[]>{
			INSTANCE;
			@Override
			public int compareTo(byte[] buffer1,int offset1,int length1,byte[] buffer2,int offset2,int length2){
				// Short circuit equal case
				if(buffer1 == buffer2 && offset1 == offset2 && length1 == length2){
					return 0;
				}
				// Bring WritableComparator code local
				int end1 = offset1 + length1;
				int end2 = offset2 + length2;
				for(int i = offset1,j = offset2;i < end1 && j < end2;i++,j++){
					int a = (buffer1[i] & 0xff);
					int b = (buffer2[j] & 0xff);
					if(a != b){
						return a - b;
					}
				}
				return length1 - length2;
			}
		}

		enum UnsafeComparer implements Comparer<byte[]>{
			INSTANCE;
			static final Unsafe theUnsafe;
			/** The offset to the first element in a byte array. */
			static final int BYTE_ARRAY_BASE_OFFSET;
			static{
				if(UNSAFE_UNALIGNED){
					theUnsafe = UnsafeAccess.theUnsafe;
				}else{
					// It doesn't matter what we throw;
					// it's swallowed in getBestComparer().
					throw new Error();
				}
				BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
				// sanity check - this should never fail
				if(theUnsafe.arrayIndexScale(byte[].class) != 1){
					throw new AssertionError();
				}
			}
			static final boolean littleEndian = ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN);

			/**
			 * Returns true if x1 is less than x2, when both values are treated as unsigned long. Both values are passed
			 * as is read by Unsafe. When platform is Little Endian, have to convert to corresponding Big Endian value
			 * and then do compare. We do all writes in Big Endian format.
			 */
			static boolean lessThanUnsignedLong(long x1,long x2){
				if(littleEndian){
					x1 = Long.reverseBytes(x1);
					x2 = Long.reverseBytes(x2);
				}
				return (x1 + Long.MIN_VALUE) < (x2 + Long.MIN_VALUE);
			}

			/**
			 * Returns true if x1 is less than x2, when both values are treated as unsigned int. Both values are passed
			 * as is read by Unsafe. When platform is Little Endian, have to convert to corresponding Big Endian value
			 * and then do compare. We do all writes in Big Endian format.
			 */
			static boolean lessThanUnsignedInt(int x1,int x2){
				if(littleEndian){
					x1 = Integer.reverseBytes(x1);
					x2 = Integer.reverseBytes(x2);
				}
				return (x1 & 0xffffffffL) < (x2 & 0xffffffffL);
			}

			/**
			 * Returns true if x1 is less than x2, when both values are treated as unsigned short. Both values are
			 * passed as is read by Unsafe. When platform is Little Endian, have to convert to corresponding Big Endian
			 * value and then do compare. We do all writes in Big Endian format.
			 */
			static boolean lessThanUnsignedShort(short x1,short x2){
				if(littleEndian){
					x1 = Short.reverseBytes(x1);
					x2 = Short.reverseBytes(x2);
				}
				return (x1 & 0xffff) < (x2 & 0xffff);
			}

			/**
			 * Checks if Unsafe is available
			 * 
			 * @return true, if available, false - otherwise
			 */
			public static boolean isAvailable(){
				return theUnsafe != null;
			}

			/**
			 * Lexicographically compare two arrays.
			 *
			 * @param buffer1 left operand
			 * @param buffer2 right operand
			 * @param offset1 Where to start comparing in the left buffer
			 * @param offset2 Where to start comparing in the right buffer
			 * @param length1 How much to compare from the left buffer
			 * @param length2 How much to compare from the right buffer
			 * @return 0 if equal, < 0 if left is less than right, etc.
			 */
			@Override
			public int compareTo(byte[] buffer1,int offset1,int length1,byte[] buffer2,int offset2,int length2){
				// Short circuit equal case
				if(buffer1 == buffer2 && offset1 == offset2 && length1 == length2){
					return 0;
				}
				final int minLength = Math.min(length1,length2);
				final int minWords = minLength / SIZEOF_LONG;
				final long offset1Adj = offset1 + BYTE_ARRAY_BASE_OFFSET;
				final long offset2Adj = offset2 + BYTE_ARRAY_BASE_OFFSET;
				/*
				 * Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a time is no slower than comparing
				 * 4 bytes at a time even on 32-bit. On the other hand, it is substantially faster on 64-bit.
				 */
				// This is the end offset of long parts.
				int j = minWords << 3; // Same as minWords * SIZEOF_LONG
				for(int i = 0;i < j;i += SIZEOF_LONG){
					long lw = theUnsafe.getLong(buffer1,offset1Adj + (long)i);
					long rw = theUnsafe.getLong(buffer2,offset2Adj + (long)i);
					long diff = lw ^ rw;
					if(diff != 0){
						return lessThanUnsignedLong(lw,rw) ? -1 : 1;
					}
				}
				int offset = j;
				if(minLength - offset >= SIZEOF_INT){
					int il = theUnsafe.getInt(buffer1,offset1Adj + offset);
					int ir = theUnsafe.getInt(buffer2,offset2Adj + offset);
					if(il != ir){
						return lessThanUnsignedInt(il,ir) ? -1 : 1;
					}
					offset += SIZEOF_INT;
				}
				if(minLength - offset >= SIZEOF_SHORT){
					short sl = theUnsafe.getShort(buffer1,offset1Adj + offset);
					short sr = theUnsafe.getShort(buffer2,offset2Adj + offset);
					if(sl != sr){
						return lessThanUnsignedShort(sl,sr) ? -1 : 1;
					}
					offset += SIZEOF_SHORT;
				}
				if(minLength - offset == 1){
					int a = (buffer1[(int)(offset1 + offset)] & 0xff);
					int b = (buffer2[(int)(offset2 + offset)] & 0xff);
					if(a != b){
						return a - b;
					}
				}
				return length1 - length2;
			}
		}
	}

	/**
	 * @param left left operand
	 * @param right right operand
	 * @return True if equal
	 */
	public static boolean equals(final byte[] left,final byte[] right){
		// Could use Arrays.equals?
		// noinspection SimplifiableConditionalExpression
		if(left == right)
			return true;
		if(left == null || right == null)
			return false;
		if(left.length != right.length)
			return false;
		if(left.length == 0)
			return true;
		// Since we're often comparing adjacent sorted data,
		// it's usual to have equal arrays except for the very last byte
		// so check that first
		if(left[left.length - 1] != right[right.length - 1])
			return false;
		return compareTo(left,right) == 0;
	}

	public static boolean equals(final byte[] left,int leftOffset,int leftLen,final byte[] right,int rightOffset,int rightLen){
		// short circuit case
		if(left == right && leftOffset == rightOffset && leftLen == rightLen){
			return true;
		}
		// different lengths fast check
		if(leftLen != rightLen){
			return false;
		}
		if(leftLen == 0){
			return true;
		}
		// Since we're often comparing adjacent sorted data,
		// it's usual to have equal arrays except for the very last byte
		// so check that first
		if(left[leftOffset + leftLen - 1] != right[rightOffset + rightLen - 1])
			return false;
		return LexicographicalComparerHolder.BEST_COMPARER.compareTo(left,leftOffset,leftLen,right,rightOffset,rightLen) == 0;
	}

	/**
	 * @param a left operand
	 * @param buf right operand
	 * @return True if equal
	 */
	public static boolean equals(byte[] a,ByteBuffer buf){
		if(a == null)
			return buf == null;
		if(buf == null)
			return false;
		if(a.length != buf.remaining())
			return false;
		// Thou shalt not modify the original byte buffer in what should be read
		// only operations.
		ByteBuffer b = buf.duplicate();
		for(byte anA:a){
			if(anA != b.get()){
				return false;
			}
		}
		return true;
	}

	/**
	 * Return true if the byte array on the right is a prefix of the byte array on the left.
	 */
	public static boolean startsWith(byte[] bytes,byte[] prefix){
		return bytes != null && prefix != null && bytes.length >= prefix.length
				&& LexicographicalComparerHolder.BEST_COMPARER.compareTo(bytes,0,prefix.length,prefix,0,prefix.length) == 0;
	}

	/**
	 * @param a lower half
	 * @param b upper half
	 * @return New array that has a in lower half and b in upper half.
	 */
	public static byte[] add(final byte[] a,final byte[] b){
		return add(a,b,EMPTY_BYTE_ARRAY);
	}

	/**
	 * @param a first third
	 * @param b second third
	 * @param c third third
	 * @return New array made from a, b and c
	 */
	public static byte[] add(final byte[] a,final byte[] b,final byte[] c){
		byte[] result = new byte[a.length + b.length + c.length];
		System.arraycopy(a,0,result,0,a.length);
		System.arraycopy(b,0,result,a.length,b.length);
		System.arraycopy(c,0,result,a.length + b.length,c.length);
		return result;
	}

	/**
	 * @param arrays all the arrays to concatenate together.
	 * @return New array made from the concatenation of the given arrays.
	 */
	public static byte[] add(final byte[][] arrays){
		int length = 0;
		for(int i = 0;i < arrays.length;i++){
			length += arrays[i].length;
		}
		byte[] result = new byte[length];
		int index = 0;
		for(int i = 0;i < arrays.length;i++){
			System.arraycopy(arrays[i],0,result,index,arrays[i].length);
			index += arrays[i].length;
		}
		return result;
	}

	/**
	 * @param a array
	 * @param length amount of bytes to grab
	 * @return First <code>length</code> bytes from <code>a</code>
	 */
	public static byte[] head(final byte[] a,final int length){
		if(a.length < length){
			return null;
		}
		byte[] result = new byte[length];
		System.arraycopy(a,0,result,0,length);
		return result;
	}

	/**
	 * @param a array
	 * @param length amount of bytes to snarf
	 * @return Last <code>length</code> bytes from <code>a</code>
	 */
	public static byte[] tail(final byte[] a,final int length){
		if(a.length < length){
			return null;
		}
		byte[] result = new byte[length];
		System.arraycopy(a,a.length - length,result,0,length);
		return result;
	}

	/**
	 * @param a array
	 * @param length new array size
	 * @return Value in <code>a</code> plus <code>length</code> prepended 0 bytes
	 */
	public static byte[] padHead(final byte[] a,final int length){
		byte[] padding = new byte[length];
		for(int i = 0;i < length;i++){
			padding[i] = 0;
		}
		return add(padding,a);
	}

	/**
	 * @param a array
	 * @param length new array size
	 * @return Value in <code>a</code> plus <code>length</code> appended 0 bytes
	 */
	public static byte[] padTail(final byte[] a,final int length){
		byte[] padding = new byte[length];
		for(int i = 0;i < length;i++){
			padding[i] = 0;
		}
		return add(a,padding);
	}

	/**
	 * Split passed range. Expensive operation relatively. Uses BigInteger math. Useful splitting ranges for MapReduce
	 * jobs.
	 * 
	 * @param a Beginning of range
	 * @param b End of range
	 * @param num Number of times to split range. Pass 1 if you want to split the range in two; i.e. one split.
	 * @return Array of dividing values
	 */
	public static byte[][] split(final byte[] a,final byte[] b,final int num){
		return split(a,b,false,num);
	}

	/**
	 * Split passed range. Expensive operation relatively. Uses BigInteger math. Useful splitting ranges for MapReduce
	 * jobs.
	 * 
	 * @param a Beginning of range
	 * @param b End of range
	 * @param inclusive Whether the end of range is prefix-inclusive or is considered an exclusive boundary. Automatic
	 *            splits are generally exclusive and manual splits with an explicit range utilize an inclusive end of
	 *            range.
	 * @param num Number of times to split range. Pass 1 if you want to split the range in two; i.e. one split.
	 * @return Array of dividing values
	 */
	public static byte[][] split(final byte[] a,final byte[] b,boolean inclusive,final int num){
		byte[][] ret = new byte[num + 2][];
		int i = 0;
		Iterable<byte[]> iter = iterateOnSplits(a,b,inclusive,num);
		if(iter == null)
			return null;
		for(byte[] elem:iter){
			ret[i++] = elem;
		}
		return ret;
	}

	/**
	 * Iterate over keys within the passed range, splitting at an [a,b) boundary.
	 */
	public static Iterable<byte[]> iterateOnSplits(final byte[] a,final byte[] b,final int num){
		return iterateOnSplits(a,b,false,num);
	}

	/**
	 * Iterate over keys within the passed range.
	 */
	public static Iterable<byte[]> iterateOnSplits(final byte[] a,final byte[] b,boolean inclusive,final int num){
		byte[] aPadded;
		byte[] bPadded;
		if(a.length < b.length){
			aPadded = padTail(a,b.length - a.length);
			bPadded = b;
		}else if(b.length < a.length){
			aPadded = a;
			bPadded = padTail(b,a.length - b.length);
		}else{
			aPadded = a;
			bPadded = b;
		}
		if(compareTo(aPadded,bPadded) >= 0){
			throw new IllegalArgumentException("b <= a");
		}
		if(num <= 0){
			throw new IllegalArgumentException("num cannot be <= 0");
		}
		byte[] prependHeader = {1,0};
		final BigInteger startBI = new BigInteger(add(prependHeader,aPadded));
		final BigInteger stopBI = new BigInteger(add(prependHeader,bPadded));
		BigInteger diffBI = stopBI.subtract(startBI);
		if(inclusive){
			diffBI = diffBI.add(BigInteger.ONE);
		}
		final BigInteger splitsBI = BigInteger.valueOf(num + 1);
		// when diffBI < splitBI, use an additional byte to increase diffBI
		if(diffBI.compareTo(splitsBI) < 0){
			byte[] aPaddedAdditional = new byte[aPadded.length + 1];
			byte[] bPaddedAdditional = new byte[bPadded.length + 1];
			for(int i = 0;i < aPadded.length;i++){
				aPaddedAdditional[i] = aPadded[i];
			}
			for(int j = 0;j < bPadded.length;j++){
				bPaddedAdditional[j] = bPadded[j];
			}
			aPaddedAdditional[aPadded.length] = 0;
			bPaddedAdditional[bPadded.length] = 0;
			return iterateOnSplits(aPaddedAdditional,bPaddedAdditional,inclusive,num);
		}
		final BigInteger intervalBI;
		try{
			intervalBI = diffBI.divide(splitsBI);
		}catch(Exception e){
			// LOG.error("Exception caught during division", e);
			logger.log(Level.SEVERE,"Exception caught during division",e);
			return null;
		}
		final Iterator<byte[]> iterator = new Iterator<byte[]>(){
			private int i = -1;

			@Override
			public boolean hasNext(){
				return i < num + 1;
			}

			@Override
			public byte[] next(){
				i++;
				if(i == 0)
					return a;
				if(i == num + 1)
					return b;
				BigInteger curBI = startBI.add(intervalBI.multiply(BigInteger.valueOf(i)));
				byte[] padded = curBI.toByteArray();
				if(padded[1] == 0)
					padded = tail(padded,padded.length - 2);
				else
					padded = tail(padded,padded.length - 1);
				return padded;
			}

			@Override
			public void remove(){
				throw new UnsupportedOperationException();
			}
		};
		return new Iterable<byte[]>(){
			@Override
			public Iterator<byte[]> iterator(){
				return iterator;
			}
		};
	}

	/**
	 * @param bytes array to hash
	 * @param offset offset to start from
	 * @param length length to hash
	 */
	public static int hashCode(byte[] bytes,int offset,int length){
		int hash = 1;
		for(int i = offset;i < offset + length;i++)
			hash = (31 * hash) + (int)bytes[i];
		return hash;
	}

	/**
	 * @param t operands
	 * @return Array of byte arrays made from passed array of Text
	 */
	public static byte[][] toByteArrays(final String[] t){
		byte[][] result = new byte[t.length][];
		for(int i = 0;i < t.length;i++){
			result[i] = toBytes(t[i]);
		}
		return result;
	}

	/**
	 * @param t operands
	 * @return Array of binary byte arrays made from passed array of binary strings
	 */
	public static byte[][] toBinaryByteArrays(final String[] t){
		byte[][] result = new byte[t.length][];
		for(int i = 0;i < t.length;i++){
			result[i] = toBytesBinary(t[i]);
		}
		return result;
	}

	/**
	 * @param column operand
	 * @return A byte array of a byte array where first and only entry is <code>column</code>
	 */
	public static byte[][] toByteArrays(final String column){
		return toByteArrays(toBytes(column));
	}

	/**
	 * @param column operand
	 * @return A byte array of a byte array where first and only entry is <code>column</code>
	 */
	public static byte[][] toByteArrays(final byte[] column){
		byte[][] result = new byte[1][];
		result[0] = column;
		return result;
	}
}

class UnsafeAvailChecker{
	static final Logger logger = Logger.getLogger(UnsafeAvailChecker.class.getName());
	private static boolean avail = false;
	private static boolean unaligned = false;
	static{
		avail = AccessController.doPrivileged(new PrivilegedAction<Boolean>(){
			@Override
			public Boolean run(){
				try{
					@SuppressWarnings("restriction")
					Field f = Unsafe.class.getDeclaredField("theUnsafe");
					f.setAccessible(true);
					return f.get(null) != null;
				}catch(Throwable e){
					logger.log(Level.WARNING,"sun.misc.Unsafe is not available/accessible",e);
				}
				return false;
			}
		});
		// 判断是否有权限使用
		if(avail){
			try{
				Class<?> clazz = Class.forName("java.nio.Bits");
				Method m = clazz.getDeclaredMethod("unaligned");
				m.setAccessible(true);
				unaligned = (Boolean)m.invoke(null);
			}catch(Exception e){
				logger.log(Level.WARNING,"java.nio.Bits#unaligned() check failed." + "Unsafe based read/write of primitive types won't be used",e);
			}
		}
	}

	/**
	 * @return true when running JVM is having sun's Unsafe package available in it and underlying system having
	 *         unaligned-access capability.
	 */
	public static boolean unaligned(){
		return unaligned;
	}
}

@SuppressWarnings("restriction")
final class UnsafeAccess{
	static final Logger logger = Logger.getLogger(UnsafeAccess.class.getName());
	public static final Unsafe theUnsafe;
	static{
		theUnsafe = (Unsafe)AccessController.doPrivileged(new PrivilegedAction<Object>(){
			@Override
			public Object run(){
				try{
					Field f = Unsafe.class.getDeclaredField("theUnsafe");
					f.setAccessible(true);
					return f.get(null);
				}catch(Throwable e){
					logger.log(Level.WARNING,"sun.misc.Unsafe is not accessible",e);
				}
				return null;
			}
		});
	}
}
